package db;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public abstract class Connector
{
    class ConnectionStore
    {
        public final Connection connection;
        public final long lastAccess;

        public ConnectionStore(Connection connection)
        {
            this.connection = connection;
            this.lastAccess = System.currentTimeMillis();
        }
    }

    private Queue<ConnectionStore> pool;
    private int timeout;
    private int idleOpenTime;
    private int selectTimeoutSec;
    private int updateTimeoutSec;
    private int poolSize;
    private String name;
    private HashSet<Connection> connectionTasks;
    private long counter = 0;

    public int getSelectTimeout()
    {
        return selectTimeoutSec;
    }

    public int getUpdateTimeout()
    {
        return updateTimeoutSec;
    }

    public Connector(String name, int timeout, int poolSize, int idleOpenTime,
                     int selectTimeoutSec, int updateTimeoutSec, int samplingInterval)
    {
        this.connectionTasks = new HashSet<Connection>();
        this.name = name;
        this.timeout = timeout;
        this.idleOpenTime = idleOpenTime;
        this.selectTimeoutSec = selectTimeoutSec;
        this.updateTimeoutSec = updateTimeoutSec;
        this.poolSize = poolSize;
        this.pool = new LinkedList<ConnectionStore>();
    }

    public Connector(String name, int timeout, int poolSize, int idleOpenTime,
                     int selectTimeoutSec, int updateTimeoutSec)
    {
        this(name, timeout, poolSize, idleOpenTime, selectTimeoutSec,
                        updateTimeoutSec, 10000);
    }

    public Connection waitForConnection() throws Exception
    {
        return waitForConnectionTask();
    }

    public String getName()
    {
        return name;
    }

    Connection createConnectionTask(Connection connection)
    {
        connectionTasks.add(connection);
        return connection;
    }

    AtomicLong idleOpenMeter = new AtomicLong();
    AtomicLong totalWaitMeter = new AtomicLong();
    AtomicInteger activeMeter = new AtomicInteger();
    AtomicLong waitingMeter = new AtomicLong();;

    Connection waitForConnectionTask() throws TimeoutException, SQLException
    {
        SQLException createConnectionException = null;
        SQLException firstException = null;
        totalWaitMeter.getAndIncrement();
        for (int i = 0; i < 2; i++)
        {
            Connection connection = null;
            synchronized (this)
            {
                if (pool.isEmpty() == false)
                {
                    ConnectionStore storedConn = pool.remove();
                    connection = storedConn.connection;
                    if (System.currentTimeMillis() - storedConn.lastAccess < idleOpenTime)
                    {
                        try
                        {
                            if (connection.isClosed() == false)
                            {
                                return createConnectionTask(connection);
                            }
                        }
                        catch (SQLException e)
                        {
                            if (firstException == null)
                            {
                                firstException = e;
                            }
                        }
                    }
                    else
                    {
                        idleOpenMeter.getAndIncrement();
                    }
                }
                else if (activeMeter.get() == poolSize)
                {
                    waitingMeter.getAndIncrement();
                    int loops = 0;
                    try
                    {
                        NotifyWaiter waiter = new NotifyWaiter(this, timeout);
                        while (pool.isEmpty())
                        {
                            if (waiter.waitForNotify() == false)
                            {
                                String message = "waitForConnection wait timeout, loops="
                                                + loops
                                                + ", name="
                                                + this.name
                                                + ", timeout=" + timeout;
                                throw new TimeoutException(message);
                            }
                            loops++;
                        }
                    }
                    finally
                    {
                        waitingMeter.decrementAndGet();
                    }

                    ConnectionStore storedConn = pool.remove();
                    connection = storedConn.connection;
                    if (System.currentTimeMillis() - storedConn.lastAccess < idleOpenTime)
                    {
                        try
                        {
                            if (connection.isClosed() == false)
                            {
                                return createConnectionTask(connection);
                            }
                        }
                        catch (SQLException e)
                        {
                            if (firstException == null)
                            {
                                firstException = e;
                            }
                        }
                    }
                    else
                    {
                        idleOpenMeter.getAndIncrement();
                    }
                }
            }

            boolean newActiveConnection = connection == null;
            if (connection != null)
            {
                try
                {
                    connection.close();
                }
                catch (SQLException e)
                {
                }
            }

            try
            {
                connection = createConnection();
            }
            catch (SQLException e)
            {
                connection = null;
                if (createConnectionException == null)
                {
                    // The first exception is more important because it is the
                    // first sign of
                    // something failing.
                    createConnectionException = e;
                }
            }
            if (connection != null)
            {
                try
                {
                    if (connection.isClosed() == false)
                    {
                        synchronized (this)
                        {
                            if (newActiveConnection)
                            {
                                if (activeMeter.get() < poolSize)
                                {
                                    activeMeter.getAndIncrement();
                                    return createConnectionTask(connection);
                                }
                            }
                            else
                            {
                                return createConnectionTask(connection);
                            }
                        }
                    }
                    else
                    {
                        throw new SQLException(
                                        "Created a new DB connection but it is closed.");
                    }
                }
                catch (SQLException e)
                {
                    if (firstException == null)
                    {
                        firstException = e;
                    }
                }
                try
                {
                    connection.close();
                }
                catch (SQLException e)
                {
                    if (firstException == null)
                    {
                        firstException = e;
                    }
                }
            }
            else
            {
                if (newActiveConnection == false)
                {
                    activeMeter.getAndDecrement();
                }
            }
        }

        String message = "waitForConnectionTask failed: Cannot connect to database "
                        + name;
        if (createConnectionException != null)
        {
            throw new SQLException(message, createConnectionException);
        }
        else if (firstException != null)
        {
            throw new SQLException(message, firstException);
        }
        else
        {
            throw new SQLException(message);
        }
    }

    public synchronized void releaseConnection(Connection connection)
    {
        boolean flag = connectionTasks.contains(connection);
        if (flag)
        {
            connectionTasks.remove(connection);
            pool.add(new ConnectionStore(connection));
            if (waitingMeter.get() > 0)
            {
                notify();
            }
        }
        else
        {
            System.err.println("Connector.releaseConnection inconsistency");
        }
    }

    protected void initializePool() throws SQLException
    {
        for (int i = 0; i < poolSize; i++)
        {
            Connection connection = createConnection();
            if (connection == null)
            {
                System.err.println("Cannot create initial connections");
                return;
            }
            if (connection.isClosed() == false)
            {
                activeMeter.getAndIncrement();
                pool.add(new ConnectionStore(connection));
            }
            else
            {
                // This is OK. The DB may be down. When the DB is up
                // waitForConnection will succeed
                // creating the connections
                System.err.println("Cannot open initial connections");
                connection.close();
            }
        }
    }

    public HashMap<Integer, Integer> executeCall(String categoryOverride, String sql,
                    Object... objects) throws Exception
    {
        Accessor acc = new Accessor(this);
        try
        {
            return acc.executeCall(categoryOverride, sql, objects);
        }
        finally
        {
            acc.close();
        }
    }

    public int executeUpdateCall(String categoryOverride, String sql,
                    Object... objects) throws Exception
    {
        Accessor acc = new Accessor(this);
        try
        {
            return acc.executeUpdateCall(categoryOverride, sql, objects);
        }
        finally
        {
            acc.close();
        }
    }

    public RowSet execute(String categoryOverride, String sql,
                    Object... objects) throws Exception
    {
        Accessor acc = new Accessor(this);
        try
        {
            return acc.execute(categoryOverride, sql, objects);
        }
        finally
        {
            acc.close();
        }
    }

    public int executeUpdate(String categoryOverride, String sql,
                    Object... objects) throws Exception
    {
        Accessor acc = new Accessor(this);
        try
        {
            return acc.executeUpdate(categoryOverride, sql, objects);
        }
        finally
        {
            acc.close();
        }
    }

    public RowSet executeUpdateAndGetGeneratedKeys(String categoryOverride,
                    String sql, Object... objects) throws Exception
    {
        Accessor acc = new Accessor(this);
        try
        {
            return acc.executeUpdateAndGetGeneratedKeys(categoryOverride, sql,
                objects);
        }
        finally
        {
            acc.close();
        }
    }

    public int[] executeBatchUpdate(String categoryOverride, String sql,
                    Object[][] objects) throws Exception
    {
        Accessor acc = new Accessor(this);
        try
        {
            return acc.executeBatchUpdate(categoryOverride, sql, objects);
        }
        finally
        {
            acc.close();
        }
    }

    public int[] executeBatch(String categoryOverride, String... sqls)
                    throws Exception
    {
        Accessor acc = new Accessor(this);
        try
        {
            return acc.executeBatch(categoryOverride, sqls);
        }
        finally
        {
            acc.close();
        }
    }

    protected abstract Connection createConnection() throws SQLException;

    public static void main(String[] args) throws Exception
    {
        String database = "mytest";
        String host = "localhost";
        int port = 3306;
        String username = "eric";
        String password = "eric123";
        int period = 30000;
        MySqlConnector connector = new MySqlConnector(database, host, port,
                        database, null, username, password, 1, 10000, 100000, 0, 0,
                        period * 10);

        connector.execute("", "insert my_test(id,name,message,my_testcol) values(?,?,?,?)", new Object[] {Integer.valueOf(1), "2333", "abc" ,"aaaaa"});

    }
}

